package com.gengzp.mq.client.processor;

import com.gengzp.mq.client.annotations.MqListener;
import com.gengzp.mq.client.core.MqCoreFunc;
import com.gengzp.mq.client.core.MqGlobalVariableManager;
import com.gengzp.mq.client.core.MqMessageConsumeInterface;
import com.gengzp.mq.client.model.message.MqMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

import static com.gengzp.mq.client.constants.MqThreadNameConstants.MQ_CONSUMER_THREAD_NAME_PREFIX;
import static com.gengzp.mq.client.constants.MqTimeoutConstants.ACK_MESSAGE_CONNECT_TIMEOUT;
import static com.gengzp.mq.client.constants.MqTimeoutConstants.SUBSCRIBE_TOPIC_WAIT_TIMEOUT;

/**
 * @ClassName MqConsumerBeanProcessor
 * @Description mq消费者 bean 处理器
 * @Author gengzp
 * @Date 2025/9/4 17:41
 */
@Component
public class MqConsumerBeanProcessor implements BeanPostProcessor {

    private static final Logger logger = LoggerFactory.getLogger(MqConsumerBeanProcessor.class);

    @Autowired
    private MqCoreFunc mqCoreFunc;

    @Autowired
    private MqGlobalVariableManager mqGlobalVariableManager;

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        // 只处理带有 @MqListener 注解的bean
        MqListener mqListenerAnnotation = bean.getClass().getAnnotation(MqListener.class);
        if (mqListenerAnnotation == null) {
            return bean;
        }

        // 获取当前订阅的主题
        String topic = mqListenerAnnotation.topic();
        if (topic == null || topic.isBlank()) {
            logger.error("消息队列监听bean:{} 使用了@MqListener 注解, 必须填写 topic 主题", beanName);
            throw new RuntimeException("消息队列监听bean: " + beanName + " 使用了@MqListener 注解注册成为监听器, 必须填写 topic 主题");
        }

        // 检查 bean 是否实现了 MqMessageConsumeInterface 接口
        if (!(bean instanceof MqMessageConsumeInterface)) {
            logger.error("消息队列监听bean:{} 使用了@MqListener 注解, 必须实现 MqMessageConsumeInterface 接口", beanName);
            throw new RuntimeException("消息队列监听bean: " + beanName + " 使用了@MqListener 注解注册成为监听器, 必须实现 MqMessageConsumeInterface 接口");
        }

        // 检查 bean 是否使用了 @Lazy 懒加载注解
        if (bean.getClass().isAnnotationPresent(Lazy.class)) {
            logger.error("消息队列监听bean: {} 禁止使用 @Lazy 注解(懒加载), 否则无法注册监听器", beanName);
            throw new RuntimeException(String.format("消息队列监听bean: %s 禁止使用 @Lazy 注解(懒加载), 否则无法注册监听器", beanName));
        }

        // 创建一个线程, 对 topic 进行订阅
        Thread subscribeThread = new Thread(() -> {
            while (true) {
                // 消息
                MqMessage message = null;
                do {
                    try {
                        message = mqCoreFunc.subscribe(topic);
                    } catch (Exception e) {
                        logger.error("订阅主题:{} 异常:{}", topic, e.getMessage());
                        try {
                            Thread.sleep(SUBSCRIBE_TOPIC_WAIT_TIMEOUT);
                        } catch (Exception ignored) {
                        }
                    }
                } while (message == null);

                // 消息消费成功标识
                boolean consumeSuccessFlag = false;
                try {
                    logger.info("============== 获取到消息, 开始消费, 消息id: {} ==============",
                            message.getMessageId());
                    ((MqMessageConsumeInterface) bean).consumeMessage(message);
                    consumeSuccessFlag = true;
                    logger.info("============== 消费成功, 消息id: {} ==============",
                            message.getMessageId());
                } catch (Exception e) {
                    logger.error("============== 消费失败, 消息id: {}, 异常信息: {} ==============",
                            message.getMessageId(), e.getMessage(), e);
                }

                // 消息确认
                if (consumeSuccessFlag) {
                    boolean ackSuccessFlag = false;
                    int retryCount = 0;
                    while (!ackSuccessFlag && retryCount < 3) {
                        try {
                            mqCoreFunc.ackMessage(message.getMessageId());
                            ackSuccessFlag = true;
                        } catch (Exception e) {
                            retryCount++;
                            logger.error("消息id: {} Ack确认失败, 错误信息: {}", message.getMessageId(), e.getMessage(), e);
                            try {
                                Thread.sleep(ACK_MESSAGE_CONNECT_TIMEOUT);
                            } catch (Exception ignored) {
                            }
                        }
                    }
                }
            }
        }, MQ_CONSUMER_THREAD_NAME_PREFIX + topic);
        // 设置为守护线程
        subscribeThread.setDaemon(true);

        // 注册监听器到全局变量, 若注册失败表示订阅当前 topic 的消息监听器已经被注册过
        if (!mqGlobalVariableManager.registerTopicListener(topic, beanName, subscribeThread)) {
            String existBeanName = mqGlobalVariableManager.getMqListenerBeanNameByTopic(topic);
            logger.error("同一个服务实例对同一个topic只能有一个监听器, 当前topic: {} 已存在监听器: {}, 因此 {} 无法注册成为监听器",
                    topic, existBeanName, beanName);
            throw new RuntimeException(String.format("同一个服务实例对同一个topic只能有一个监听器, 当前topic: %s 已存在监听器: %s, 因此 %s 无法注册成为监听器",
                    topic, existBeanName, beanName));
        }

        return bean;
    }

}
